package com.bbx.iotdb.iotdbdemo.mqtt;

import com.alibaba.fastjson.JSON;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.lang.reflect.Array;
import java.sql.BatchUpdateException;
import java.util.ArrayList;
import java.util.List;
import java.util.zip.ZipEntry;

@Slf4j
@RestController
@RequestMapping("/mqttClient")
public class MQTTClient {




    /**
     * 消息有效负载示例
     * {
     * "device":"root.sg.d1",
     * "timestamp":1586076045524,
     * "measurements":["s1","s2"],
     * "values":[0.530635,0.530635]
     * }
     * 或者
     * {
     * "device":"root.sg.d1",
     * "timestamps":[1586076045524,1586076065526],
     * "measurements":["s1","s2"],
     * "values":[[0.530635,0.530635], [0.530655,0.530695]]
     * }
     *
     * @param size
     */
    @RequestMapping("/sendMsg")
    public void sendMsg(int size) {
        List<Mqtt3AsyncClient> mqtt3AsyncClients = new ArrayList<>();



        long timestamp = 21812L;

        List<String> measurements = new ArrayList<>();
        measurements.add("status");
        measurements.add("temperature");

        List<List<Object>> values = new ArrayList<>();
        List<Object> value1 = new ArrayList<>();

        value1.add(false);
        value1.add(2.11);
        values.add(value1);

        for (int i = 0; i < size; i++) {
            Mqtt3AsyncClient client = Mqtt3Client.builder()
                    .identifier("aaaaaaaaaa" + i)
                    .serverHost("192.168.10.85")
                    .serverPort(2883)
                    .buildAsync();
            mqtt3AsyncClients.add(client);
        }

        log.info("mqtt3AsyncClients.size[{}]", mqtt3AsyncClients.size());


        for (int i = 0; i < mqtt3AsyncClients.size(); i++) {
            PayloadEntity payloadEntity = new PayloadEntity();
            List<Long> timeStamps = new ArrayList<>();
            timeStamps.add(timestamp+i);
            payloadEntity.setDevice("root.ddl.wf06.wt06");
            payloadEntity.setTimestamps(timeStamps);
            payloadEntity.setMeasurements(measurements);
            payloadEntity.setValues(values);
            String s1 = JSON.toJSONString(payloadEntity);
            Mqtt3AsyncClient client = mqtt3AsyncClients.get(i);
            client.connectWith()
                    .simpleAuth()
                    .username("root")
                    .password("root".getBytes())
                    .applySimpleAuth()
                    .send()
                    .whenComplete((connAck, throwable) -> {
                        if (throwable != null) {
                            //处理连接失败
                            log.info("连接失败[{}]",s1);
                        } else {
                            log.info("[{}]", "连接成功");
                            client.publishWith()
                                    .topic("ddl/topic")
                                    .qos(MqttQos.AT_LEAST_ONCE)
                                    .payload(s1.getBytes())
                                    //保留标识：由代理保存制定主题的最后一条消息，新的client 订阅该主题时，将会接收到这条消息
                                    .retain(true).send();

                            client.disconnect();
                        }
                    });
        }
    }


}
